package org.sup.hi;

import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URL;

import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.fs.FsUrlStreamHandlerFactory;
import org.apache.zookeeper.ZooKeeper;
import org.jgroups.Address;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.View;
import org.sup.hi.SyncPrimitive.Barrier;

public class BarrierChat extends ReceiverAdapter{
	JChannel channel;
	Address address;
	static ZooKeeper zk = null;
    static Integer mutex;
	String user_name = System.getProperty("user.name","n/a");
	
	BarrierChat(){
	}
	
	private void start(String adress) throws Exception{
		channel = new JChannel();
		
		channel.setReceiver(this);
		channel.connect("ChatCluster");
		Barrier barrier = new Barrier(adress,"/b1",3);
		barrier.enter();
		eventLoop(adress);
		channel.close();
	}
	
	private void eventLoop(String adress) {
		BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
		while(true){
			try{
				System.out.print("> ");
				System.out.flush();
				String line = in.readLine().toLowerCase();
				if(line.startsWith("quit") || line.startsWith("exit"))
					break;
				// read files
				String command[] = new String[2];
				String strMsg = null;
				command = line.split("\\s");
				if(command[0].equals("read")){
					// Read the file from HDFS and send the content
					InputStream is = null;
					ByteArrayOutputStream byteArrayOS = new ByteArrayOutputStream();
					is = new URL(command[1]).openStream();
					IOUtils.copyBytes(is, byteArrayOS, 4096, false);
					strMsg = byteArrayOS.toString();
					// clean up
					byteArrayOS.close();
					is.close();
				}else if(command[0].equals("send")){
					// Send the message
					strMsg = command[1];
				}
				strMsg = "[" + user_name + "]" + strMsg;
				Message msg = new Message(null, null, strMsg);
				channel.send(msg);
			}catch(Exception e){
				System.out.print(e.getMessage());
			}
		}
	}
	
	@Override
	public void receive(Message msg) {
		System.out.println(msg.getSrc()+": "+msg.getObject());
	}

	// Called when a new instance joins the cluster
	// or an existing instance leaves
	@Override
	public void viewAccepted(View view) {
		System.out.println("** view: "+ view);
	}
	
	static {
		URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
	}
	
	public static void main(String[] args) throws Exception {
		String adress = args[0];
		
		new BarrierChat().start(adress);
	}
}
